from pyflink.common import SimpleStringSchema, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer

env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)



# set_bootstrap_servers： broker列表
# set_topics： 消费的topic
# set_group_id: 消费者组
# set_starting_offsets：开启读取数据的位置,latest:读取最新的数据，earliest：从最早读取数据
# set_value_only_deserializer： 数据格式

kafka_source = KafkaSource.builder()\
    .set_bootstrap_servers('master:9092')\
    .set_topics('bigdata').set_group_id('my-group')\
    .set_starting_offsets(KafkaOffsetsInitializer.earliest())\
    .set_value_only_deserializer(SimpleStringSchema())\
    .build()

kafka_ds = env.from_source(kafka_source, WatermarkStrategy.no_watermarks(), 'Kafka Source')

kafka_ds.print()

env.execute()